package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.model.user.UserInfo;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.reopsitory.AlbumIndexRepository;
import com.atguigu.tingshu.search.reopsitory.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"all"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;
    @Autowired
    private CategoryFeignClient categoryFeignClient;
    @Autowired
    private UserInfoFeignClient userFeignClient;
    @Autowired
    private AlbumIndexRepository albumIndexRepository;
    @Autowired
    private SuggestIndexRepository suggestIndexRepository;
    @Autowired
    private ThreadPoolExecutor executor;
    @Autowired
    private ElasticsearchClient elasticsearchClient;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RedissonClient redissonClient;
    @Override
    public void upperAlbum(Long albumId) {

        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
        //远程调用albumService服务获取albumInfo数据
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture =CompletableFuture.supplyAsync(()->{
            Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);
            AlbumInfo albumInfo = albumInfoResult.getData();
            Assert.notNull(albumInfo,"专辑为空");
            BeanUtils.copyProperties(albumInfo,albumInfoIndex);
            albumInfoIndex.setIsFinished(String.valueOf(albumInfo.getIsFinished()));
            return albumInfo;
        },executor);
        //远程调用userService服务获取用户数据
        CompletableFuture<Void> userCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<UserInfo> userInfoResult = userFeignClient.getUserInfo(albumInfo.getUserId());
            UserInfo userInfo = userInfoResult.getData();
            Assert.notNull(albumInfo, "用户为空");
            albumInfoIndex.setAnnouncerName(userInfo.getNickname());
        },executor);
        //远程调用categoryService服务获取分类数据
        CompletableFuture<Void> categoryCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<BaseCategoryView> baseCategoryViewResult = categoryFeignClient.getCategory(albumInfo.getCategory3Id());
            BaseCategoryView baseCategoryView = baseCategoryViewResult.getData();
            Assert.notNull(baseCategoryView, "分类为空");
            albumInfoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(baseCategoryView.getCategory2Id());
        },executor);
        //远程调用albumService服务获取专辑状态数据
        CompletableFuture<Void> albumStatCompletableFuture = CompletableFuture.runAsync(() -> {
            Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumStat(albumId);
            AlbumStatVo albumStatVo = albumStatVoResult.getData();
            Assert.notNull(albumStatVo, "专辑状态为空");
            albumInfoIndex.setPlayStatNum(albumStatVo.getPlayStatNum());
            albumInfoIndex.setSubscribeStatNum(albumStatVo.getSubscribeStatNum());
            albumInfoIndex.setBuyStatNum(albumStatVo.getBuyStatNum());
            albumInfoIndex.setCommentStatNum(albumStatVo.getCommentStatNum());
        },executor);
        //设置商品热度
        albumInfoIndex.setHotScore(new Random().nextDouble(1000d));
        //添加专辑属性值
        CompletableFuture<Void> attributeValueCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            List<AttributeValueIndex> attributeValueIndexList = albumInfo.getAlbumAttributeValueVoList().stream().map(albumAttributeValueVo -> {
                AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                attributeValueIndex.setAttributeId(albumAttributeValueVo.getAttributeId());
                attributeValueIndex.setValueId(albumAttributeValueVo.getValueId());
                return attributeValueIndex;
            }).toList();
            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
        },executor);
        CompletableFuture.allOf(albumInfoCompletableFuture,
                userCompletableFuture,
                categoryCompletableFuture,
                albumStatCompletableFuture,
                attributeValueCompletableFuture).join();
        //保存专辑信息到es中
        albumIndexRepository.save(albumInfoIndex);
        //保存专辑id到布隆过滤器中
        RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(albumId);
        //上架添加提示词
        //专辑标题提示
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(UUID.randomUUID().toString().replace("-", ""));
        suggestIndex.setTitle(albumInfoIndex.getAlbumTitle());
        suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()}));
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())}));
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())}));
        suggestIndexRepository.save(suggestIndex);
        //专辑简介提示
        suggestIndex = new SuggestIndex();
        suggestIndex.setId(UUID.randomUUID().toString().replace("-", ""));
        suggestIndex.setTitle(albumInfoIndex.getAlbumIntro());
        suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumIntro()}));
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumIntro())}));
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumIntro())}));
        suggestIndexRepository.save(suggestIndex);
        //专辑作者提示
        suggestIndex = new SuggestIndex();
        suggestIndex.setId(UUID.randomUUID().toString().replace("-", ""));
        suggestIndex.setTitle(albumInfoIndex.getAnnouncerName());
        suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAnnouncerName()}));
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAnnouncerName())}));
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAnnouncerName())}));
        suggestIndexRepository.save(suggestIndex);
    }

    @Override
    public void lowerAlbum(Long albumId) {
        albumIndexRepository.deleteById(albumId);
        suggestIndexRepository.deleteById(albumId);

    }

    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery albumIndexQuery) {
        //创建查询结果对象
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        //编写动态dsl语句
        SearchRequest searchRequest = queryBuildDsl(albumIndexQuery);
        //得到查询结果
        SearchResponse<AlbumInfoIndex> searchResponse = null;
        try {
            searchResponse = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        //将结果放入返回结果集中
        Integer pageNo = albumIndexQuery.getPageNo();
        albumSearchResponseVo.setPageNo(pageNo);
        Integer pageSize = albumIndexQuery.getPageSize();
        albumSearchResponseVo.setPageSize(pageSize);
        long total = searchResponse.hits().total().value();
        albumSearchResponseVo.setTotal(total);
        Long totalPage = total % pageSize == 0 ? total / pageSize : total / pageSize + 1;
        albumSearchResponseVo.setTotalPages(totalPage);
        List<AlbumInfoIndexVo> albumInfoIndexVoList = parseAlbumInfoIndexVoList(searchResponse);
        albumSearchResponseVo.setList(albumInfoIndexVoList);
        //返回结果
        return albumSearchResponseVo;
    }

    @Override
    public List<Map<String, Object>> channel(Long category1Id) {
        //远程调用albumService服务获取category3List数据
        Result<List<BaseCategory3>> category3ListResult = albumInfoFeignClient.findTopBaseCategory3(category1Id);
        Assert.notNull(category3ListResult, "三级分类为空");
        List<BaseCategory3> category3List = category3ListResult.getData();
        Map<Long, BaseCategory3> category3Map = category3List.stream().collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));
        //获取category3IdList
        List<FieldValue> category3IdList = category3List.stream()
                .map(baseCategory3 -> FieldValue.of(baseCategory3.getId())).collect(Collectors.toList());
        //编写dsl语句，查询专辑列表
        SearchResponse<AlbumInfoIndex> response = null;
        try {
            response = elasticsearchClient.search(s -> s.index("albuminfo")
                            .query(q -> q.bool(b -> b.filter(f -> f.terms(t -> t.field("category3Id")
                                    .terms(t1 -> t1.value(category3IdList))))))
                            .aggregations("agg_category3Id", ag -> ag.terms(t -> t
                                            .field("category3Id").size(7))
                                    .aggregations("agg_albumInfoByCategory3Id", agg -> agg
                                            .topHits(to -> to.size(6).sort(so -> so.
                                                    field(f -> f.field("hotScore").order(SortOrder.Desc))))))
                    , AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        Assert.notNull(response, "查询结果为空");
        //获取专辑列表
        Aggregate aggCategory3Id = response.aggregations().get("agg_category3Id");
        //获取专辑列表，并存入map中
        List<Map<String, Object>> channelList = aggCategory3Id.lterms().buckets().array().stream().map(bucket -> {
            List<AlbumInfoIndex> albumInfoIndexList = bucket.aggregations().get("agg_albumInfoByCategory3Id").topHits().hits().hits().stream().map(hit -> {
                JsonData source = hit.source();
                AlbumInfoIndex albumInfoIndex = JSON.parseObject(source.toString(), AlbumInfoIndex.class);
                return albumInfoIndex;
            }).collect(Collectors.toList());
            Map<String, Object> map = new HashMap<>();
            map.put("list", albumInfoIndexList);
            map.put("baseCategory3", category3Map.get(bucket.key()));
            return map;
        }).collect(Collectors.toList());
        return channelList;
    }

    @Override
    public List<String> completeSuggest(String keyword) {
        //编写dsl语句，查询自动补全列表
        SearchResponse<SuggestIndex> search = null;
        try {
            search = elasticsearchClient.search(s -> s.index("suggestinfo").suggest(su -> su
                    .suggesters("keyword", s1 -> s1.prefix(keyword)
                            .completion(c -> c.field("keyword").size(10)
                                    .skipDuplicates(true).fuzzy(f -> f.fuzziness("auto"))))
                    .suggesters("keywordPinyin", s1 -> s1.prefix(keyword)
                            .completion(c -> c.field("keywordPinyin").size(10)
                                    .skipDuplicates(true).fuzzy(f -> f.fuzziness("auto"))))
                    .suggesters("keywordSequence", s1 -> s1.prefix(keyword)
                            .completion(c -> c.field("keywordSequence").size(10)
                                    .skipDuplicates(true).fuzzy(f -> f.fuzziness("auto"))))
            ), SuggestIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        //获取自动补全列表并返回
        List<String> list1 = getSuggestList(search, "keyword");
        List<String> list2 = getSuggestList(search, "keywordPinyin");
        List<String> list3 = getSuggestList(search, "keywordSequence");
        List<String> list = new ArrayList<>(list1.size()+list2.size()+list3.size());
        list.addAll(list1);
        list.addAll(list2);
        list.addAll(list3);
        return list;
    }

    @Override
    public void updateLatelyAlbumRanking() {
        //获取所有一级分类id
        Result<List<Long>> category1IdListResult = categoryFeignClient.getCategory1Id();
        Assert.notNull(category1IdListResult, "一级分类为空");
        List<Long> category1IdList = category1IdListResult.getData();
        Assert.notNull(category1IdList, "一级分类为空");
        for (Long category1Id : category1IdList) {
            String[] dimensions = {"hotScore","playStatNum","subscribeStatNum","buyStatNum","commentStatNum"};
            for (String dimension : dimensions) {
                //编写dsl语句，查询专辑列表
                SearchResponse<AlbumInfoIndex> search = null;
                try {
                    search = elasticsearchClient.search(s -> s.index("albuminfo")
                                    .query(q -> q.bool(b -> b.filter(f -> f.term(t -> t.field("category1Id").value(category1Id)))))
                                    .sort(so -> so.field(f -> f.field(dimension).order(SortOrder.Desc)))
                            , AlbumInfoIndex.class);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                //专辑排序列表
                List<AlbumInfoIndex> infoIndexList = search.hits().hits().stream().map(hit -> hit.source()).collect(Collectors.toList());
                //放入缓存中
                BoundHashOperations<String,String,List<AlbumInfoIndex>> boundHashOperations = redisTemplate.boundHashOps(RedisConstant.RANKING_KEY_PREFIX + category1Id);
                boundHashOperations.put(dimension, infoIndexList);
            }
        }
    }

    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String dimension) {
        //从缓存中获取排行榜
        BoundHashOperations<String,String,List<AlbumInfoIndex>> boundHashOperations = redisTemplate.boundHashOps(RedisConstant.RANKING_KEY_PREFIX + category1Id);
        List<AlbumInfoIndex> albumInfoIndexList = boundHashOperations.get(dimension);
        if (!CollectionUtils.isEmpty(albumInfoIndexList)){
            List<AlbumInfoIndexVo> albumInfoIndexVoList = albumInfoIndexList.stream().map(albumInfoIndex -> {
                AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
                BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
                return albumInfoIndexVo;
            }).collect(Collectors.toList());
            return albumInfoIndexVoList;
        }
        return new ArrayList<>();
    }

    /**
     * 获取自动补全列表
     * @param search
     * @param keyword
     * @return
     */
    private List<String> getSuggestList(SearchResponse<SuggestIndex> search, String keyword) {
        List<String> list = new ArrayList<>();
        search.suggest().get(keyword).stream().forEach(s -> {
            List<String> stringList = s.completion().options().stream().map(o -> o.source().getTitle()).collect(Collectors.toList());
            list.addAll(stringList);
        });
        return list;
    }

    private List<AlbumInfoIndexVo> parseAlbumInfoIndexVoList(SearchResponse<AlbumInfoIndex> searchResponse) {
        List<Hit<AlbumInfoIndex>> albumInfoIndexList = searchResponse.hits().hits();
        List<AlbumInfoIndexVo> albumInfoIndexVoList = new ArrayList<>(albumInfoIndexList.size());
        for (Hit<AlbumInfoIndex> hit : albumInfoIndexList) {
            AlbumInfoIndex albumInfoIndex = hit.source();
            AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
            BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
            if (!CollectionUtils.isEmpty(hit.highlight())){
                albumInfoIndexVo.setAlbumTitle(hit.highlight().get("albumTitle").get(0));
            }
            albumInfoIndexVoList.add(albumInfoIndexVo);
        }
        return albumInfoIndexVoList;
    }

    private SearchRequest queryBuildDsl(AlbumIndexQuery albumIndexQuery) {
        SearchRequest.Builder searchRequestBuilder = new SearchRequest.Builder();
        searchRequestBuilder.index("albuminfo");
        BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
        String ksyWord = albumIndexQuery.getKeyword();
        //如果关键字不为空，进行关键字查询
        if (!StringUtils.isEmpty(ksyWord)){
            boolQueryBuilder.should(s -> s.match(m -> m.field("albumTitle").query(ksyWord)))
                    .should(s -> s.match(m -> m.field("albumIntro").query(ksyWord)))
                    .should(s -> s.match(m -> m.field("announcerName").query(ksyWord)));
            //高亮
            searchRequestBuilder.highlight(h -> h.fields("albumTitle",hf ->
                    hf.preTags("<span style=color:red>").postTags("</span>")));
        }
        //根据分类id进行查询
        if (albumIndexQuery.getCategory1Id() != null){
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category1Id").value(albumIndexQuery.getCategory1Id())));
        }
        if (albumIndexQuery.getCategory2Id() != null){
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category2Id").value(albumIndexQuery.getCategory2Id())));
        }
        if (albumIndexQuery.getCategory3Id() != null){
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category3Id").value(albumIndexQuery.getCategory3Id())));
        }
        //根据属性与属性值进行过滤
        if (!CollectionUtils.isEmpty(albumIndexQuery.getAttributeList())){
            for (String attr : albumIndexQuery.getAttributeList()) {
                String[] split = attr.split(":");
                if (split != null && split.length == 2){
                    String attributeId = split[0];
                    String valueId = split[1];
                    boolQueryBuilder.filter(f -> f.nested(n -> n.path("attributeValueIndexList")
                            .query(q -> q.bool(b -> b.filter(f1 -> f1.term(t -> t.field("attributeValueIndexList.attributeId")
                                    .value(attributeId)))
                                    .filter(f2 -> f2.term(t -> t.field("attributeValueIndexList.valueId")
                                            .value(valueId)))))));
                }
            }
        }
        //根据order属性进行排序
        if (!StringUtils.isEmpty(albumIndexQuery.getOrder())){
            String[] split = albumIndexQuery.getOrder().split(":");
            String order = split[0];
            String sort = split[1];
            if ("1".equals(order)){
                searchRequestBuilder.sort(s -> s.field(f -> f.field("hotScore").order("desc".equals(sort)? SortOrder.Desc:SortOrder.Asc)));
            }
            if ("2".equals(order)){
                searchRequestBuilder.sort(s -> s.field(f -> f.field("playStatNum").order("desc".equals(sort)? SortOrder.Desc:SortOrder.Asc)));
            }
            if ("3".equals(order)){
                searchRequestBuilder.sort(s -> s.field(f -> f.field("createTime").order("desc".equals(sort)? SortOrder.Desc:SortOrder.Asc)));
            }
        }
        //分页
        searchRequestBuilder.from((albumIndexQuery.getPageNo() -1)* albumIndexQuery.getPageSize()).size(albumIndexQuery.getPageSize());
        searchRequestBuilder.query(boolQueryBuilder.build()._toQuery());
        SearchRequest searchRequest = searchRequestBuilder.build();
        System.out.println(searchRequest.toString());
        return searchRequest;
    }
}
